package gossip.v1;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public  class MembershipTable {
    private final Map<String, MemberState> members = new ConcurrentHashMap<>();
    private final MemberState localState;

    public MembershipTable(int port) {
        this.localState = new MemberState("node-" + port, new InetSocketAddress("localhost", port));
        this.members.put(localState.getNodeId(), localState); // 关键修复：初始化时加入本地节点
    }


    public Map<String, Long> getMembersDigest() {
        Map<String, Long> digest = new HashMap<>();
        getMembers().forEach((id, state) -> digest.put(id, state.getHeartbeat()));
        return digest;
    }


    public void merge(MemberState remote) {
        if (remote == null || remote.getNodeId().equals(localState.getNodeId())) {
            return;
        }

        members.compute(remote.getNodeId(), (id, local) -> {
            if (local == null) {
                System.out.printf("[%s] Added new node: %s@%s:%d%n",
                        localState.getNodeId(),
                        remote.getNodeId(),
                        remote.getAddress().getHostString(),
                        remote.getAddress().getPort());
                return remote;
            }
            // 使用向量时钟解决冲突
            if (remote.getHeartbeat() > local.getHeartbeat()) {
                return remote; // 采用新状态
            } else if (remote.getHeartbeat() == local.getHeartbeat()) {
                // 相同版本号，合并元数据
                Map<String, String> merged = new HashMap<>(local.getMetadata());
                remote.getMetadata().forEach((k, v) -> {
                    if (!merged.containsKey(k) || v.compareTo(merged.get(k)) > 0) {
                        merged.put(k, v);
                    }
                });
                local.setMetadata(merged);
            }
            return local;
        }); // 保留本地状态
    }

    public void incrementLocalHeartbeat() {
        localState.incrementHeartbeat();
        // 确保更新成员表中的本地状态
        members.put(localState.getNodeId(), localState);
    }

    public void updateLocalMetadata(String key, String value) {
        localState.getMetadata().put(key, value);
        incrementLocalHeartbeat();
    }

    public Map<String, MemberState> getMembers() {
        return Collections.unmodifiableMap(members);
    }

    public MemberState getLocalState() {
        return localState;
    }

    public List<MemberState> selectRandomTargets(int count) {
        List<MemberState> candidates = new ArrayList<>(getMembers().values());
        candidates.remove(localState); // 排除自身

        // 优先选择长时间未通信的节点
        candidates.sort(Comparator.comparingLong(MemberState::getUpdateTime));

        int targetCount = Math.min(count, candidates.size());
        return candidates.subList(0, targetCount);
    }
}
